package com.yang.thread;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.Random;

/**
 * 生产者消费者模式示例
 * 演示使用BlockingQueue实现线程间的协作
 */
public class ProducerConsumerExample {
    
    public static void main(String[] args) {
        System.out.println("=== 生产者消费者模式示例 ===");
        
        // 使用有界队列
        demonstrateBoundedQueue();
        
        // 使用无界队列
        demonstrateUnboundedQueue();
    }
    
    /**
     * 有界队列示例 - 当队列满时生产者会阻塞
     */
    private static void demonstrateBoundedQueue() {
        System.out.println("\n--- 有界队列示例 (容量: 5) ---");
        
        // 创建容量为5的有界队列
        BlockingQueue<Product> queue = new ArrayBlockingQueue<>(5);
        
        // 创建生产者线程
        Thread producer1 = new Thread(new Producer(queue, "生产者1", 8), "Producer-1");
        Thread producer2 = new Thread(new Producer(queue, "生产者2", 6), "Producer-2");
        
        // 创建消费者线程
        Thread consumer1 = new Thread(new Consumer(queue, "消费者1"), "Consumer-1");
        Thread consumer2 = new Thread(new Consumer(queue, "消费者2"), "Consumer-2");
        Thread consumer3 = new Thread(new Consumer(queue, "消费者3"), "Consumer-3");
        
        // 启动所有线程
        producer1.start();
        producer2.start();
        consumer1.start();
        consumer2.start();
        consumer3.start();
        
        // 等待生产者完成
        try {
            producer1.join();
            producer2.join();
            
            // 等待一段时间让消费者处理完剩余产品
            Thread.sleep(3000);
            
            // 中断消费者线程
            consumer1.interrupt();
            consumer2.interrupt();
            consumer3.interrupt();
            
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        System.out.println("有界队列示例结束，剩余产品数量: " + queue.size());
    }
    
    /**
     * 无界队列示例
     */
    private static void demonstrateUnboundedQueue() {
        System.out.println("\n--- 无界队列示例 ---");
        
        // 创建无界队列
        BlockingQueue<Product> queue = new LinkedBlockingQueue<>();
        
        // 创建快速生产者和慢速消费者
        Thread fastProducer = new Thread(new FastProducer(queue, "快速生产者", 10), "FastProducer");
        Thread slowConsumer = new Thread(new SlowConsumer(queue, "慢速消费者"), "SlowConsumer");
        
        fastProducer.start();
        slowConsumer.start();
        
        try {
            fastProducer.join();
            Thread.sleep(2000); // 让消费者处理一些产品
            slowConsumer.interrupt();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        System.out.println("无界队列示例结束，剩余产品数量: " + queue.size());
    }
}

/**
 * 产品类
 */
class Product {
    private final int id;
    private final String name;
    private final long createTime;
    
    public Product(int id, String name) {
        this.id = id;
        this.name = name;
        this.createTime = System.currentTimeMillis();
    }
    
    public int getId() {
        return id;
    }
    
    public String getName() {
        return name;
    }
    
    public long getCreateTime() {
        return createTime;
    }
    
    @Override
    public String toString() {
        return String.format("Product{id=%d, name='%s'}", id, name);
    }
}

/**
 * 生产者类
 */
class Producer implements Runnable {
    private final BlockingQueue<Product> queue;
    private final String producerName;
    private final int productCount;
    private final Random random = new Random();
    
    public Producer(BlockingQueue<Product> queue, String producerName, int productCount) {
        this.queue = queue;
        this.producerName = producerName;
        this.productCount = productCount;
    }
    
    @Override
    public void run() {
        try {
            for (int i = 1; i <= productCount; i++) {
                // 创建产品
                Product product = new Product(i, producerName + "-产品" + i);
                
                // 放入队列（如果队列满了会阻塞）
                System.out.println(producerName + " 准备生产: " + product + 
                                 " [队列大小: " + queue.size() + "]");
                
                queue.put(product); // 阻塞方法
                
                System.out.println(producerName + " 已生产: " + product + 
                                 " [队列大小: " + queue.size() + "]");
                
                // 模拟生产时间
                Thread.sleep(500 + random.nextInt(1000));
            }
            
            System.out.println(producerName + " 生产完毕，共生产 " + productCount + " 个产品");
            
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println(producerName + " 被中断");
        }
    }
}

/**
 * 消费者类
 */
class Consumer implements Runnable {
    private final BlockingQueue<Product> queue;
    private final String consumerName;
    private final Random random = new Random();
    
    public Consumer(BlockingQueue<Product> queue, String consumerName) {
        this.queue = queue;
        this.consumerName = consumerName;
    }
    
    @Override
    public void run() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                // 从队列中取出产品（如果队列空了会阻塞）
                Product product = queue.take(); // 阻塞方法
                
                System.out.println(consumerName + " 开始消费: " + product + 
                                 " [队列大小: " + queue.size() + "]");
                
                // 模拟消费时间
                Thread.sleep(800 + random.nextInt(1200));
                
                long processTime = System.currentTimeMillis() - product.getCreateTime();
                System.out.println(consumerName + " 消费完成: " + product + 
                                 " [处理时间: " + processTime + "ms]");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println(consumerName + " 被中断，停止消费");
        }
    }
}

/**
 * 快速生产者 - 演示生产速度快于消费速度的情况
 */
class FastProducer implements Runnable {
    private final BlockingQueue<Product> queue;
    private final String producerName;
    private final int productCount;
    
    public FastProducer(BlockingQueue<Product> queue, String producerName, int productCount) {
        this.queue = queue;
        this.producerName = producerName;
        this.productCount = productCount;
    }
    
    @Override
    public void run() {
        try {
            for (int i = 1; i <= productCount; i++) {
                Product product = new Product(i, producerName + "-产品" + i);
                
                queue.put(product);
                System.out.println(producerName + " 快速生产: " + product + 
                                 " [队列大小: " + queue.size() + "]");
                
                // 快速生产
                Thread.sleep(200);
            }
            
            System.out.println(producerName + " 快速生产完毕");
            
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println(producerName + " 被中断");
        }
    }
}

/**
 * 慢速消费者 - 演示消费速度慢于生产速度的情况
 */
class SlowConsumer implements Runnable {
    private final BlockingQueue<Product> queue;
    private final String consumerName;
    
    public SlowConsumer(BlockingQueue<Product> queue, String consumerName) {
        this.queue = queue;
        this.consumerName = consumerName;
    }
    
    @Override
    public void run() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                Product product = queue.take();
                
                System.out.println(consumerName + " 开始慢速消费: " + product + 
                                 " [队列大小: " + queue.size() + "]");
                
                // 慢速消费
                Thread.sleep(2000);
                
                System.out.println(consumerName + " 慢速消费完成: " + product);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println(consumerName + " 被中断，停止消费");
        }
    }
}
